package com.hu.flink12.api.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: hujianjun
 * @Date: 2021/2/4 0:17
 * @Describe: 自定义分区
 */
public class CustomerPartion {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //TODO 2.读取数据源
        DataStream<Long> ds = env.fromSequence(1, 30);

        //TODO 3.转换操作
//        ds.global();//都发往第一个task中
//        ds.forward();//一对一
//        ds.shuffle();
//        ds.rebalance();
//        ds.rescale();

        DataStream<Long> longDataStream = ds.partitionCustom(new Partitioner<Long>() {
            @Override
            public int partition(Long value, int i) {
                if (value % 3 == 0) {
                    return 0;
                } else if (value % 5 == 0) {
                    return 1;
                } else {
                    return 2;
                }
            }
        }, t -> t);

        //TODO 4.sink操作
        longDataStream.print();

        //TODO 5.执行程序
        env.execute();
    }
}
